package com.zhao.streaming

import com.zhao.commonUtil.{ConnectionManager, JsonParser}
import com.zhao.util.{ParamParserUtil, SparkUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.DStream

import java.sql.{Connection, Date, PreparedStatement}

/**
 * Description: 使用服务位置解析<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2021/1/6 13:40
 *
 * @author 柒柒
 * @version : 1.0
 */

object ServiceUsageInfo {
  def main(args: Array[String]): Unit = {
    Logger.getRootLogger.setLevel(Level.ERROR)
    //创建流
    val (message, ssc) = SparkUtil.createKafkaStream(args)
    //解析参数
    val paramMap: Map[String, String] = ParamParserUtil.argsToOptionMap(args)
    //过滤服务数据流
    val serviceParser: DStream[(String, Int)] = serviceUsageParser(message, paramMap).reduceByKey(_ + _)
    //持久化到数据库
    persistServiceUsage(serviceParser)
    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 持久化到数据库
   * @param dstream
   */
  def persistServiceUsage(dstream: DStream[(String, Int)]) = {
    dstream.foreachRDD(rdd =>{
      rdd.foreachPartition(iter =>{
        val now: Long = System.currentTimeMillis()
        var conn: Connection = null
        var pstmt: PreparedStatement = null
        try{
          conn = ConnectionManager.getConnnection
          conn.setAutoCommit(false)
          val sql =
            """
              |insert into fz_bigscreen_service_usage_info(stat_date,area,town,service_type,use_count,update_time)
              |values(?,?,?,?,?,?)
              |on duplicate key update use_count=values(use_count) + use_count,update_time=values(update_time)
              |""".stripMargin
          pstmt = conn.prepareStatement(sql)
          while (iter.hasNext){
            val data: (String, Int) = iter.next()
            val service_type = data._1.split("#")(0)
            val areaAndTown: Array[String] = data._1.split("#")(1).split("_")
            val area = areaAndTown(0)
            val town = areaAndTown(1)
            val use_count = data._2
            pstmt.setDate(1,new Date(now))
            pstmt.setString(2, area)
            pstmt.setString(3, town)
            pstmt.setString(4, service_type)
            pstmt.setInt(5, use_count)
            pstmt.setTimestamp(6, new java.sql.Timestamp(now))
            pstmt.addBatch()
          }
          pstmt.executeBatch()
          conn.commit()
        }catch{
          case e: Throwable => e.printStackTrace()
            conn.rollback()
        }finally {
          ConnectionManager.closeAll(null,pstmt,conn)
        }
      })
    })
  }

  /**
   * 解析服务所在位置
   * @param dstream
   * @param paramMap
   * @return
   */
  def serviceUsageParser(dstream: DStream[ConsumerRecord[String, String]], paramMap: Map[String, String]) = {
    val filterStream: DStream[ConsumerRecord[String, String]] = dstream.filter(line => {
      val info: String = line.value()
      var flag = false
      //解析json
      try {
        val jsonParser = new JsonParser(info)
        if (info.contains("event") && info.contains("service_name") && info.contains("toon_type")
          && info.contains("city") && info.contains("area") && info.contains("town")) {
          val toon_type = jsonParser.getSCJsnField("properties", "toon_type")
          val service_name = jsonParser.getSCJsnField("properties", "service_name")
          val city = jsonParser.getSCJsnField("properties", "city")
          val area = jsonParser.getSCJsnField("properties", "area")
          val town = jsonParser.getSCJsnField("properties", "town")
          if (paramMap("toon_type").equals(toon_type) && !"".equals(service_name)
            && city != null && city.contains("佛山") && area != "null" && town != "null") {
            flag = true
          }
        }
      } catch {
        case e: Throwable => e.printStackTrace()
      }
      flag
    })

    val finalStream: DStream[(String, Int)] = filterStream.map(line => {
      val info: String = line.value()
      val jsonParser = new JsonParser(info)
      val area = jsonParser.getSCJsnField("properties", "area")
      val town = jsonParser.getSCJsnField("properties", "town")
      val service_name = jsonParser.getSCJsnField("properties", "service_name")
      (service_name + "#" + area + "-" + town, 1)
    })
    finalStream
  }

}















